package parallel;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Calendar;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import event.Generator;
import event.SequenceGenerator;

public class ParallelSequenceGenerator extends Configured implements Tool {
	static int getNumberOfRegions(JobContext context) {
		Configuration conf = context.getConfiguration();
		return conf.getInt("mapred.map.tasks", 1);
	}
	static void setNumberOfsequences(Configuration conf, long numsequences) {
		conf.setLong("num-sequences", numsequences);
	}
	static void setDelimiter(Configuration conf, String delimit) {
		conf.set("mapred.textoutputformat.separator", delimit);
	}
	public static class SequenceGenMapper extends Mapper<LongWritable, NullWritable, LongWritable, Text> {
		private Generator eventGenerator = null;
		protected void setup(Context context)
		throws IOException, InterruptedException {
			// set a fake start point
			Calendar calendar = Calendar.getInstance();
			calendar.set(2000,1,1);
			BasicRangeInputSplit inputSplit = (BasicRangeInputSplit)context.getInputSplit();
			long currTime = calendar.getTimeInMillis();
			// the number of regions is the same as the mapper number
			int numOfRegions = getNumberOfRegions(context);
			eventGenerator = new SequenceGenerator(currTime,context, inputSplit, numOfRegions);
		}
		protected void map(LongWritable sequence, NullWritable ignored,
				Context context) throws IOException, 
				InterruptedException{
			eventGenerator.generate(sequence.get(), "");
		}
	}
	/**
	 * @param args the cli arguments
	 * @throws ClassNotFoundException 
	 * @throws InterruptedException 
	 * @throws URISyntaxException 
	 */
	public int run(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
		Configuration conf = HBaseConfiguration.create(getConf());
		setDelimiter(conf, ",");
		setNumberOfsequences(conf, Long.parseLong(args[0]));
		Job job = new Job(conf, "parallelEventGeneration");
		job.setMapperClass(SequenceGenMapper.class);
		job.setNumReduceTasks(0);
		job.setJarByClass(ParallelSequenceGenerator.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		job.setInputFormatClass(EventRangeInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		TextOutputFormat.setOutputPath(job, new Path(args[1]));
		return job.waitForCompletion(true)? 0: 1;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new ParallelSequenceGenerator(), args);
		System.exit(res);
	}

}
